package com.study.mq.consumer;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.study.utils.ConnectionUtils;

public class ConsumerMsg
{
    private final static String QUEUE_NAME = "queue1";
    
    public static void main(String[] argv) throws Exception
    {
        Runnable runnable = new Runnable()
        {
            @Override
            public void run()
            {
                try
                {
                    String thread_name = Thread.currentThread().getName();
                    
                    // 获取到连接
                    Connection connection = ConnectionUtils.getConnection();
                    //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
                    Channel channel = connection.createChannel();
                    
                    /*
                     * 这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。
                     * 换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。
                     * 
                     * 该设置，仅在手动ack时生效
                     * 应用场景：有两个消费者消费100条数据，一个消费慢，一个消费快。
                     *         若不设置，则每个消费者会各消费50条。设置之后，会在消费者确认ack后，才会给消费者推送消息
                     */
                    channel.basicQos(1);
                    
                    // 声明队列
                    //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
                    /**
                     * 参数明细
                     * 1、queue 队列名称
                     * 2、durable 是否持久化，如果持久化，mq重启后队列还在
                     * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
                     * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
                     * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
                     */
                    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
                    //实现消费方法
                    Consumer consumer = new DefaultConsumer(channel)
                    {
                        // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
                        /**
                         * 当接收到消息后此方法将被调用
                         * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
                         * @param envelope 信封，通过envelope
                         * @param properties 消息属性
                         * @param body 消息内容
                         * @throws IOException
                         */
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                byte[] body) throws IOException
                        {
                            try
                            {
                                //交换机
                                String exchange = envelope.getExchange();
                                //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
                                long deliveryTag = envelope.getDeliveryTag();
                                // body 即消息体
                                String msg = new String(body, "utf-8");
                                System.out.println(thread_name+" 消费消息 : " + msg + "!");
                                //Thread.sleep(1000);
                                //手动确认
                                channel.basicAck(deliveryTag,false);
                            }
                            catch (Exception e)
                            {
                                e.printStackTrace();
                            }
                            
                        }
                    };
                    
                    // 监听队列，第二个参数：是否自动进行消息确认。
                    //参数：String queue, boolean autoAck, Consumer callback
                    /**
                     * 参数明细：
                     * 1、queue 队列名称
                     * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
                     * 3、callback，消费方法，当消费者接收到消息要执行的方法
                     */
                    channel.basicConsume(QUEUE_NAME, false, consumer);
                }
                catch (IOException e)
                {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                catch (TimeoutException e)
                {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        };
        
        Thread t = new Thread(runnable);
        t.start();
    }
}
